package com.risenetsoft.elasticsearch.api;

import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

/**
 * Created by Zhanggd on 2017/8/8.
 */
public class BulkApiTest {
    //集群的名称
    private static final String CLUSTER_NAME = "elasticsearch";
    //服务IP,可以添加多个IP地址
    private static final String IP = "192.168.190.104";
    //端口：浏览器上直接访问的为9200，Java客户端访问的端口为9200
    private static final Integer PORT = 9300;

    //1.设置集群名称：默认是elasticsearch，并设置client.transport.sniff为true，
    // 使客户端嗅探整个集群状态，把集群中的其他机器IP加入到客户端中
    private static Settings settings = Settings.builder()
            .put("cluster.name",CLUSTER_NAME)
            .put("client.transport.sniff",true)
            .put("client.transport.ping_timeout","120s")
            .build();

    //设置全局私有对象
    private static TransportClient client = null;

    static {
        try {
            client = new PreBuiltTransportClient(settings)
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(IP), PORT));
        } catch (UnknownHostException e) {
            System.out.println("初始化客户端失败...");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        bulk();
    }

    public static void bulk(){
        BulkRequestBuilder bulkRequest = client.prepareBulk();

        try {
            //添加一条记录
            bulkRequest.add(client.prepareIndex("twitter","tweet","1")
                    .setSource(jsonBuilder()
                            .startObject()
                            .field("user", "mic")
                            .field("postDate", new Date())
                            .field("message", "commit message to elasticsearch")
                            .endObject()));

            //添加一条记录
            bulkRequest.add(client.prepareIndex("twitter","tweet","2")
                    .setSource(jsonBuilder()
                            .startObject()
                            .field("user","feifei")
                            .field("postDate",new Date())
                            .field("message","commit message to elasticsearch again")
                            .endObject()));

            BulkResponse bulkResponse = bulkRequest.get();
            if (bulkResponse.hasFailures()){
                BulkItemResponse[] items = bulkResponse.getItems();
                for (BulkItemResponse item:items){
                    String id = item.getId();
                    String index = item.getIndex();
                    String type = item.getType();
                    System.out.println("id:"+id+",index:"+index+",type:"+type);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
